package se.sics.kompics.tutorial.tutorial3.RIWCM;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pDeliver;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.tutorial.tutorial3.RIWCM.timeout.SleepTimeout;
import se.sics.kompics.tutorial.tutorial3.RIWCM.event.ReadRequest;
import se.sics.kompics.tutorial.tutorial3.RIWCM.event.ReadResponse;
import se.sics.kompics.tutorial.tutorial3.RIWCM.event.WriteRequest;
import se.sics.kompics.tutorial.tutorial3.RIWCM.event.WriteResponse;
import se.sics.kompics.tutorial.tutorial3.RIWCM.message.AckMessage;
import se.sics.kompics.tutorial.tutorial3.RIWCM.message.ReadMessage;
import se.sics.kompics.tutorial.tutorial3.RIWCM.message.ReadValMessage;
import se.sics.kompics.tutorial.tutorial3.RIWCM.message.WriteMessage;

public class RIWCMImpl extends ComponentDefinition {

	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	Positive<Timer> timer = requires(Timer.class);
	Negative<AtomicRegister> atReg = provides(AtomicRegister.class);
	
	private String[] commands;
	private int lastCommand;
	private static final Logger logger = LoggerFactory.getLogger(RIWCMImpl.class);
	
	private Address self;
	private Set<Address> neighbors;
	private Set<Address>[] writeSet;
	private Set<ReadSetElement>[] readSet;
	private boolean[] reading;
	private int[] reqId,  v, ts, mrank, writeVal, readVal;
	private int nbrReg;
	
	public RIWCMImpl(){
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handleReadRequest, atReg);
		subscribe(handleWriteRequest, atReg);
		subscribe(handleReadMessage, pp2p);
		subscribe(handleReadValMessage, pp2p);
		subscribe(handleWriteMessage, pp2p);
		subscribe(handleAckMessage, pp2p);
		subscribe(handleTimeout, timer);
	}
	
	// Init Handler
	Handler<RIWCMInit> handleInit = new Handler<RIWCMInit>() {
		
		@SuppressWarnings("unchecked")
		@Override
		public void handle(RIWCMInit event) {
			commands = event.getCommandScript().split(":");
			lastCommand = -1;
			
			// Topology init
			self = event.getSelf();
			neighbors = event.getNeighborSet();
			
			// State init (with the given number of registers)
			int r = event.getNbRegister();
			nbrReg = r;
			reqId = new int[r]; 
			v = new int[r];
			ts = new int[r];
			mrank = new int[r];
			writeVal = new int[r];
			readVal = new int[r];
			
			reading = new boolean[r];
			writeSet = new Set[r];
			readSet = new Set[r];
			
			for(int i = 0; i < r; i++){
				writeSet[i] = new HashSet<Address>();
				readSet[i] = new HashSet<ReadSetElement>();
				reading[i] = false;
				reqId[i] = 0;
				v[i] = 0;
				ts[i] = 0;
				mrank[i] = 0;
			}
		}
	};
	
	// Handle timeout for waking up
	Handler<SleepTimeout> handleTimeout = new Handler<SleepTimeout>(){

		@Override
		public void handle(SleepTimeout arg0) {
			doNextCommand();
		}
	};
	
	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
			doNextCommand();
		}
	};
	
	// Read request handler
	
	Handler<ReadRequest> handleReadRequest = new Handler<ReadRequest>() {

		@Override
		public void handle(ReadRequest request) {
			Integer r = request.getRegisterNumber();
			reqId[r] ++;
			reading[r] = true;
			writeSet[r].clear();
			readSet[r].clear();
			doBroadcast(new ReadMessage(self, r, reqId[r]));
			logger.info("Broadcasted ReadMessage for register {} for read request", r);
		}
	};
	
	// Write request handler
	Handler<WriteRequest> handleWriteRequest = new Handler<WriteRequest>(){

		@Override
		public void handle(WriteRequest request) {
			Integer r = request.getRegisterNumber();
			reqId[r] ++;
			writeVal[r] = request.getValue();
			readSet[r].clear();
			writeSet[r].clear();
			doBroadcast(new ReadMessage(self, r, reqId[r]));
			logger.info("Broadcasted ReadMessage for register {} for write request", r);
		}
		
	};
	
	Handler<ReadMessage> handleReadMessage = new Handler<ReadMessage>() {
		
		@Override
		public void handle(ReadMessage message) {
			Integer r = message.getRegisterNumber();
			trigger(new Pp2pSend(message.getSource(), new ReadValMessage(self,
					r, message.getReqId(), ts[r], mrank[r], v[r])), pp2p);
			logger.info("ReadMessage received, responding with the value of register "+r+" : {}", v[r]);
		}
	};
	
	Handler<ReadValMessage> handleReadValMessage = new Handler<ReadValMessage>() {
		
		@Override
		public void handle(ReadValMessage message) {
			Integer r = message.getRegisterNumber();
			if (message.getReqId().equals(reqId[r])) {
				readSet[r].add(new ReadSetElement(message.getTs(),
						message.getMrank(), message.getV(), message.getSource()));
				doCheckReadConsensus(r);
			} else {
			}
		}
	};
	
	Handler<WriteMessage> handleWriteMessage = new Handler<WriteMessage>() {
		
		@Override
		public void handle(WriteMessage message) {
			int r = message.getRegisterNumber();
			logger.info("Received write message ("+message.getTs()+", "+message.getMrank()+")  -  State : ("+ts[r]+", "+mrank[r]+")");
			if (	(message.getTs() > ts[r]) ||
					(message.getTs() == ts[r] && message.getMrank() > mrank[r])	) {
				v[r] = message.getV();
				ts[r] = message.getTs();
				mrank[r] = message.getMrank();
				logger.info("Wrting a new value in register "+r+" : {}.", v[r]);
			} 
			/*else if (message.getMrank() > mrank[r]) {
				v[r] = message.getV();
				ts[r] = message.getTs();
				mrank[r] = message.getMrank();
				logger.info("Wrting a new value in register "+r+" : {}.", v[r]);
			}*/
			trigger(new Pp2pSend(message.getSource(), new AckMessage(self, r, message.getReqid())), pp2p);
		}
	};
	
	Handler<AckMessage> handleAckMessage = new Handler<AckMessage>() {
		
		@Override
		public void handle(AckMessage message) {
			Integer r = message.getRegisterNumber();
			if (message.getReqid().equals(reqId[r])) {
				writeSet[r].add(message.getSource());
				doCheckWriteConsensus(r);
			}
		}
	};
	
	private void doCheckReadConsensus(Integer r) {
		if (readSet[r].size() > (neighbors.size()+1)/2) {
			ReadSetElement max = getMaxElement(readSet[r]);
			readVal[r] = max.getVal();
			if (reading[r]) {
				doBroadcast(new WriteMessage(self, r, reqId[r], max.getTs(), max.getMrank(), readVal[r]));
				logger.info("We are reading the value, updating the neighbors with value {}.", readVal[r]);
			} else {
				doBroadcast(new WriteMessage(self, r, reqId[r], max.getTs()+1, self.getPort(), writeVal[r]));
				logger.info("We are writing the value, updating the neighbors with value {}.", writeVal[r]);
			}
			// In order to avoid orphan messages problems
			readSet[r].clear();
		}
	}
	
	private void doCheckWriteConsensus(Integer r) {
		if (writeSet[r].size() > (neighbors.size()+1)/2) {
			if (reading[r]) {
				reading[r] = false;
				trigger(new ReadResponse(r, readVal[r]), atReg);
				logger.info("The value of register "+r+" is {}.", readVal[r]);
			} else {
				logger.info("Write command terminated");
				trigger(new WriteResponse(r), atReg);
			}
			// In order to avoid orphan messages problems
			writeSet[r].clear();
			doNextCommand();
		}
	}
	
	private final void doNextCommand() {
		lastCommand++;

		if (lastCommand > commands.length) {
			return;
		}
		if (lastCommand == commands.length) {
			logger.info("DONE ALL OPERATIONS");
			Thread applicationThread = new Thread("ApplicationThread") {
				public void run() {
					BufferedReader in = new BufferedReader(
							new InputStreamReader(System.in));
					while (true) {
						try {
							String line = in.readLine();
							doCommand(line);
						} catch (Throwable e) {
							e.printStackTrace();
						}
					}
				}
			};
			applicationThread.start();
			return;
		}
		String op = commands[lastCommand];
		doCommand(op);
	}
	
	private void doCommand(String cmd) {
		if (cmd.startsWith("R")) {
			if(cmd.length() > 1) {
				logger.info("Reading from register "+cmd.substring(1));
				Integer r = Integer.parseInt(cmd.substring(1));
				if ( r >= 0 && r < nbrReg) {
					handleReadRequest.handle(new ReadRequest(r));
				} else {
					logger.info("Invalid register number: {}. Try 'help'.", r);
					doNextCommand();
				}
			} else {
				logger.info("You have to specify a register number. Try 'help'.");
				doNextCommand();
			}
		} else if (cmd.startsWith("W")) {
			String[] parts = cmd.split("R");
			logger.info("Writing "+parts[0].substring(1)+" on register {}", parts[1]);
			handleWriteRequest.handle(new WriteRequest(Integer.parseInt(parts[1]), Integer.parseInt(parts[0].substring(1))));
		} else if (cmd.startsWith("D")) {
			logger.info("Sleeping for "+cmd.substring(1)+"ms");
			ScheduleTimeout requestCheck = new ScheduleTimeout(Integer.parseInt(cmd.substring(1)));
			requestCheck.setTimeoutEvent(new SleepTimeout(requestCheck));
			trigger(requestCheck, timer);
		} else if (cmd.equals("help")) {
			doHelp();
			doNextCommand();
		} else if (cmd.startsWith("X")) {
			doShutdown();
		} else {
			logger.info("Bad command: '{}'. Try 'help'", cmd);
			doNextCommand();
		}
	}

	private final void doHelp() {
		logger.info("Available commands: W<n>R<i>, R<i>, D<ms>, help, X");
		logger.info("WnRi: write the message 'n' on register 'i'");
		logger.info("Ri: read the register 'i'. Registers are numbered from 0 to {}.", nbrReg-1);
		logger.info("Dms: sleep for 'ms' milliseconds.");
		logger.info("help: shows this help message");
		logger.info("X: terminates this process");
	}
	
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
	
	private void doBroadcast(Pp2pDeliver message) {
		for(Address neighbor : neighbors) {
			trigger(new Pp2pSend(neighbor, message), pp2p);
		}
		trigger(new Pp2pSend(self, message), pp2p);
	}
	
	public final static ReadSetElement getMaxElement(Set<ReadSetElement> set) {
		ReadSetElement max = set.iterator().next();
		for(ReadSetElement element : set) {
			if(element.greaterThan(max)) {
				max = element;
			}
		}
		return max;
	}
		
	private class ReadSetElement {
		
		private Integer ts;
		private Integer mrank;
		private Integer val;
		private Address source;
		
		public ReadSetElement(Integer ts, Integer mrank, Integer val, Address source) {
			this.ts = ts;
			this.mrank = mrank;
			this.val = val;
			this.source = source;
		}
		
		public Integer getTs() {
			return ts;
		}
		
		public Integer getMrank() {
			return mrank;
		}
		
		public Integer getVal() {
			return val;
		}
		
		@SuppressWarnings("unused")
		public Address getSource() {
			return source;
		}
		
		public boolean greaterThan(ReadSetElement otherElement) {
			if (ts != otherElement.getTs()) {
				return (ts > otherElement.getTs());
			} else {
				return (mrank > otherElement.getMrank());
			}
		}
		
	}
	
}
